package com.gxlevi.batch_process.util

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}


object HBaseUtil {

  val conf: Configuration = HBaseConfiguration.create

  val conn: Connection = ConnectionFactory.createConnection(conf)

  val admin: Admin = conn.getAdmin

  /**
   * 获取表
   *
   * @param tableNameStr
   * @param columnFamilyName
   * @return
   */
  def getTable(tableNameStr: String, columnFamilyName: String): Table = {
    val tableName: TableName = TableName.valueOf(tableNameStr)
    if (!admin.tableExists(tableName)) {
      val tableDescriptor: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName)
      val columnFamily: ColumnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyName.getBytes()).build()
      tableDescriptor.setColumnFamily(columnFamily)
      admin.createTable(tableDescriptor.build())
    }
    conn.getTable(tableName)
  }

  /**
   * 保存单个数据
   *
   * @param tableNameStr
   * @param rowKey
   * @param columnFamilyName
   * @param columnName
   * @param columnValue
   */
  def putData(tableNameStr: String, rowKey: String, columnFamilyName: String, columnName: String, columnValue: String) = {
    val table = getTable(tableNameStr, columnFamilyName)

    try {
      val put = new Put(rowKey.getBytes())
      put.addColumn(columnFamilyName.getBytes(), columnName.getBytes(), columnValue.getBytes())
      table.put(put)
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    } finally {
      table.close()
    }
  }

  def getData(tableNameStr: String, rowKey: String, columnFamilyName: String, columnName: String) = {
    val table = getTable(tableNameStr, columnFamilyName)

    try {
      val get: Get = new Get(rowKey.getBytes())
      val result = table.get(get)
      if (null != result && result.containsColumn(columnFamilyName.getBytes, columnName.getBytes)) {
        val colValueBytes: Array[Byte] = result.getValue(columnFamilyName.getBytes, columnName.getBytes)
        Bytes.toString(colValueBytes)
      } else {
        ""
      }
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
        ""
    } finally {
      table.close()
    }
  }

  /**
   * 批量添加数据
   *
   * @param tableNameStr
   * @param rowKey
   * @param columnFamilyName
   * @param map
   */
  def putMapData(tableNameStr: String, rowKey: String, columnFamilyName: String, map: Map[String, Any]) = {
    val table = getTable(tableNameStr, columnFamilyName)

    try {
      for (elem <- map) {
        val put = new Put(rowKey.getBytes())
        put.addColumn(columnFamilyName.getBytes, elem._1.getBytes, elem._2.toString.getBytes)
        table.put(put)
      }
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
    } finally {
      table.close()
    }
  }

  def getMapData(tableNameStr: String, rowKey: String, columnFamily: String, columnList: List[String]) = {
    val table = getTable(tableNameStr, columnFamily)

    try {
      val get: Get = new Get(rowKey.getBytes())
      val result: Result = table.get(get)
      columnList.map {
        col =>
          val values: Array[Byte] = result.getValue(columnFamily.getBytes, col.getBytes)
          if (values != null && values.size > 0) {
            col -> Bytes.toString(values)
          } else {
            "" -> ""
          }
      }.filter(_._1 != "").toMap
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
        Map[String, String]()
    } finally {
      table.close()
    }
  }

  def deleteData(tableNameStr: String, rowKey: String, columnFamilyName: String) = {
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      val delete = new Delete(rowKey.getBytes())
      table.delete(delete)
    } catch {
      case ex: Exception => ex.printStackTrace()
    } finally {
      table.close()
    }
  }

  def main(args: Array[String]): Unit = {
    //    println(getTable("test","info"))
    //    putData("test","rk0001","info","t1","hello world")
    //    println(getData("test", "rk0001", "info", "t1"))
//    val map = Map(
//      "t2"->"scala",
//      "t3"->"hive",
//      "t4"->"flink"
//    )
//    putMapData("test", "rk0002", "info", map)
//    val stringToString: Map[String, String] = getMapData("test", "rk0002", "info", List("t2", "t3"))
//    println(stringToString)

    deleteData("test","rk0002","info")
  }

}
